package com.shujia.flink.cep

import com.shujia.flink.fraud.Event
import org.apache.flink.api.scala._
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object FraudDetector {
  val MIN_AMOUNT: Double = 1.00
  val MAX_AMOUNT: Double = 500.00
  val ONE_MINUTE: Long = 10 * 1000L
}

object Demo5FraudDetector {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val lines = env.socketTextStream("node1", 8888)

    val events = lines.map(line => {
      val split = line.split(",")
      val id = split(0)
      val amount = split(1).toDouble
      Event(id, amount)
    })


    val pattern = Pattern
      .begin[Event]("begin")
      .where(event => event.amount < FraudDetector.MIN_AMOUNT)

      .next("next")
      .where(event => event.amount > FraudDetector.MAX_AMOUNT)
      //必须在执行时间完成
      .within(Time.milliseconds(FraudDetector.ONE_MINUTE))


    val patternDS = CEP.pattern(events.keyBy(_.id), pattern)


    patternDS.select(map => {
      val begin = map.getOrElse("begin", null)
      val next = map.getOrElse("next", null)
      println(begin + "\t" + next)
    })

    env.execute()


  }

  case class Event(id: String, amount: Double)

}
